#include <netinet/ip.h>

#include <stdio.h>
#include <stdint.h>
#include <signal.h>
#include <string.h>
#include <stdlib.h>
#include <stdint.h>
#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/syscall.h>
#include <sys/epoll.h>
#include <sys/prctl.h>
#include <fcntl.h>
#include <ctype.h>

#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <net/if.h>
#include <unistd.h>
#include <errno.h>


#include "hecp.h"
#include "stream_main.h"
#include "xeLOG_internal.h"

#define MAX_CH_SUPPORT 10
#define MAX_EPOLL_FD      512
#define EPOLL_TIMEOUT     1000

static struct cha_priv_struct dvr_cha_priv[MAX_CH_SUPPORT] = {{0}};



typedef struct {
	int sockfd;
	int chid;
	int sub;
	int m;
} ep_data;

ep_data epd_main[MAX_CH_SUPPORT]={{0}};

static int epollfd[8] = {0};

static char TID_main[MAX_CH_SUPPORT][64] = {{0}};
static char TID_subA[MAX_CH_SUPPORT][64] = {{0}};


static int listen_port_base    = 27000;
static int listen_port_end     = 28000;
static int listen_port_current = 27000;

static int __soe_sockfd_auto_bind(int sockfd) {
	struct sockaddr_in addr;

	addr.sin_family=AF_INET;
	addr.sin_addr.s_addr=INADDR_ANY;

	listen_port_current++;

	while (1) {
		if (listen_port_current>listen_port_end)
			listen_port_current=listen_port_base;

		addr.sin_port=htons(listen_port_current);
		if (bind(sockfd, (struct sockaddr*)&addr, sizeof(addr))==0)
			break;

		xeLOG_INFO("bind(%d) FAIL, %s",listen_port_current,strerror(errno));

		listen_port_current++;

	}
	return listen_port_current;
}

extern int _TCP_keepalive_on(int sockfd);
static int __soe_set_address_TCP_PORT( struct cha_priv_struct *hd_p )
{
	int IP0=0, IP1=0, IP2=0, IP3=0, retval=-1;
	char IPs[32]="";
	int port=0;
	int recvbuflen = (1<<20);
	struct sockaddr_in *saip=NULL;

	retval=sscanf(hd_p->address, "%d.%d.%d.%d:%d", &IP0, &IP1, &IP2, &IP3, &port);
	if (retval!=5) {
		/*unknown address format.*/
		xeLOG_ERR("invalid address %s.", hd_p->address);
		return -1;
	}
	__snprintf(IPs, "%d.%d.%d.%d", IP0, IP1, IP2, IP3);
	saip=(struct sockaddr_in *)&hd_p->addr_peer;
	if ( !inet_aton(IPs, &saip->sin_addr) ) {
		//invalid IP address
		xeLOG_ERR("invalid address %s.",IPs);
		return -1;
	}
	saip->sin_family = AF_INET;
	saip->sin_port   = htons(port);

	hd_p->send_sockfd=socket(AF_INET,SOCK_STREAM,0);
	if (hd_p->send_sockfd==-1) {
		xeLOG_ERR("socket FAIL, %s",strerror(errno));
		return -1;
	}

    if(setsockopt(hd_p->send_sockfd,SOL_SOCKET,SO_RCVBUF,(void*)&recvbuflen,sizeof(int))< 0){
        recvbuflen = CONFIG_XESOE_SNDBUF_SIZE;
        setsockopt(hd_p->send_sockfd,SOL_SOCKET,SO_RCVBUFFORCE,(void*)&recvbuflen,sizeof(int));
    }
	_TCP_keepalive_on(hd_p->send_sockfd);

	saip=(struct sockaddr_in*)&hd_p->addr_local;
	saip->sin_port = htons(__soe_sockfd_auto_bind(hd_p->send_sockfd));
	xeLOG_NOTICE("tcp local port:%d",saip->sin_port);
	retval=connect(hd_p->send_sockfd, (struct sockaddr*)&hd_p->addr_peer,sizeof(struct sockaddr_in));
	if (retval==-1) {
		close(hd_p->send_sockfd);
		hd_p->send_sockfd=-1;
		xeLOG_ERR("connect(%s:%d) FAIL, %s",IPs,port,strerror(errno));
		return -1;
	}

	fcntl(hd_p->send_sockfd, F_SETFL, fcntl(hd_p->send_sockfd, F_GETFL, 0) | O_NONBLOCK); 
	return 0;
}


int hecp_stream_open_tcp(char *hIPaddr, int chanNum, int stream, int lport, char *mode, char *addr)
{
	char url[64] = {0};
	int cid,retval, key_len, val_len;
	struct timeval tout={10,0};


	if ( stream ) {
		snprintf(url, 63, "/soe/channel%d_subA", 0);
	} else {
		snprintf(url, 63, "/soe/channel%d", 0);
	}

	cid = hecp_cli_open_TCP(hIPaddr, 9822);
	if(cid == -1){
		xeLOG_DEBUG("ch-%d, open TCP->%s:%d fail.", chanNum, hIPaddr, 9822);
		return -1;
	}
	hecp_cli_set_request_head_method(cid,HECP_METHOD_GET);
	hecp_cli_set_request_head_URL(cid, url);

	hecp_cli_add_request_entity(cid, "PROTO", "TCP");
	hecp_cli_add_request_entity(cid, "MODE", "PASSIVE");

	retval = hecp_cli_transmit(cid,&tout);
	if(retval == -1){
		xeLOG_DEBUG("ch-%d, transmit between %s:9822 fail.", chanNum, hIPaddr);
		hecp_cli_close(cid);
		return -1;
	}
	retval = hecp_cli_get_response_head_status_code(cid);
	if(retval == STATUS_CODE_200){
		hecp_cli_get_response_entity_first(cid);
		while (1) {
			char key[32] = {0};
			char val[128] = {0};
			key_len = sizeof(key);
			val_len = sizeof(val);
			retval  = hecp_cli_get_response_entity_next(cid,key,&key_len,val,&val_len);
			if(retval == -1)
                break;/*EOF*/
            STD_print("key:%s value:%s",key,val);
			if ( strcmp(key, "ADDRESS") == 0 ) {
                sprintf( mode, "%s", "PORT");
                sprintf( addr, "%s", val );
                
			} else if ( !strcmp( key, "TID" ) ){
				if ( stream ){
					memset( TID_subA[chanNum], 0, 64 );
					strncpy( TID_subA[chanNum], val, 64-1 );
				} else {
					memset( TID_main[chanNum], 0, 64 );
					strncpy( TID_main[chanNum], val, 64-1 );
				}
			}
		}
		retval = 0;
	}else{
		xeLOG_DEBUG("ch-%d, bad response code %d from %s:9822.", chanNum, retval, hIPaddr);
		retval = -1;
	}

	hecp_cli_close(cid);
	return retval;
}
int hecp_stream_close( char *hIPaddr, int chanNum, int stream )
{
	int cid, retval;
	char url[64] = {0};
	struct timeval tout = { 3, 0 };


	if ( chanNum < 0 || chanNum >= MAX_CH_SUPPORT ) {
		return -1;
	}


    if ( strnlen( TID_main[chanNum], 64 ) < 10 ) {
        memset( TID_main[chanNum], 0, 64 );
        return 0;
    }
    snprintf(url, 63, "/soe/channel%d", 0);


	cid=hecp_cli_open_TCP(hIPaddr, 9822);
	if(cid==-1){
		xeLOG_DEBUG("ch-%d, open TCP->%s:%d fail.", chanNum, hIPaddr, 9822);
		return -1;
	}
	hecp_cli_set_request_head_method(cid,HECP_METHOD_DELETE);
	hecp_cli_set_request_head_URL(cid, url);

	if ( stream )
		hecp_cli_add_request_entity(cid, "TID", TID_subA[chanNum]);
	else
		hecp_cli_add_request_entity(cid, "TID", TID_main[chanNum]);

	retval=hecp_cli_transmit(cid, &tout);
	if(retval==-1){
		xeLOG_DEBUG("ch-%d, transmit between %s:9822 fail.", chanNum, hIPaddr);
		goto exit_177;
	}

	retval=hecp_cli_get_response_head_status_code(cid);
	if ( retval == STATUS_CODE_200 ) {
		if ( stream ){
			xeLOG_INFO( "ch-%d, %s closed.", chanNum, TID_subA[chanNum] );
		} else{
			xeLOG_INFO( "ch-%d, %s closed.", chanNum, TID_main[chanNum] );
		}
	} else {
		if ( stream ){
			xeLOG_INFO( "ch-%d, %s close fail.", chanNum, TID_subA[chanNum] );
		} else{
			xeLOG_INFO( "ch-%d, %s close fail.", chanNum, TID_main[chanNum] );
		}
	}
	retval = 0;

exit_177:
	hecp_cli_close(cid);

	if ( stream ){
		memset( TID_subA[chanNum], 0, 64 );
	} else{
		memset( TID_main[chanNum], 0, 64 );
	}

	return retval;
}

int tcp_stream_test_start(char *ip)
{
    int ret;
    struct cha_priv_struct *hd_p  = &dvr_cha_priv[0];
    int channel = 0;
    int stream = 0;
    int port = 0;
    ep_data *epd;
    struct epoll_event ev;

    ret = hecp_stream_open_tcp(ip, channel, stream, port, hd_p->mode,hd_p->address);
    if(ret == -1){
        xeLOG_ERR("_ft_task_set_address_TCP_PORT FAIL");
        hd_p->error = 5;
        return -1;    
    }
    ret = __soe_set_address_TCP_PORT(hd_p);
    if(ret == -1) {
        xeLOG_ERR("_ft_task_set_address_TCP_PORT FAIL");
        hd_p->error = 6;
        return -1;
    }
    hd_p->pFd = fopen("./stream.ps","wb+");
    if(!hd_p->pFd){
        xeLOG_ERR("open FAIL");
        hd_p->error = 6;
        return -1;    
    }
    /* add tcp send_sockfd to epoll */
    memset(&ev,0x0,sizeof(ev));

    epd = &epd_main[0];

    epd->sockfd = hd_p->send_sockfd;
    epd->chid   = 0;
    ev.data.ptr = epd;
    ev.events = EPOLLIN;//level triggle
    epoll_ctl(epollfd[0],EPOLL_CTL_ADD, hd_p->send_sockfd,&ev);

    return 0;
}
int tcp_stream_test_close(struct cha_priv_struct *hd_p )
{
    epoll_ctl(epollfd[0], EPOLL_CTL_DEL, hd_p->send_sockfd, NULL);
	// close(hd_p->send_sockfd);
	// hd_p->send_sockfd = -1;

	hecp_stream_close("192.168.157.203", 0, 0 );
    if(hd_p->pFd){
        fclose(hd_p->pFd);
    }
	if(-1 != hd_p->send_sockfd ){
		close(hd_p->send_sockfd);
		hd_p->send_sockfd = -1;
	}	
    return 0;
}
#define buf_data_size (1*1024*1024)
char buffer[buf_data_size];
static int produce_data(struct cha_priv_struct *hd_p)
{
    int ret;
    socklen_t	len = sizeof(struct sockaddr);
    while(1){
        ret = recvfrom(hd_p->send_sockfd, buffer, 
            buf_data_size, MSG_NOSIGNAL|MSG_DONTWAIT, (struct sockaddr*)&hd_p->addr_peer, &len);
        if(ret > 0){
            int len = 0;
            STD_print("receive size,ret:%d",ret);
            if(hd_p->pFd){
               len =  fwrite ((void *)buffer, 1, ret, hd_p->pFd);
               if(len != ret){
                   STD_print("receive size,ret:%d len:%d",ret,len);
               }
            }
        }else{
            if ( errno == EAGAIN || errno == EWOULDBLOCK ){
                STD_print("receive block");
                break;
            }
            xeLOG_ERR("CH-%d: read fail,%s.close",hd_p->ChaID , strerror(errno));
            if(hd_p->send_sockfd != -1) {
                close(hd_p->send_sockfd);
                hd_p->send_sockfd = -1;
            }
            return -1;
        }
    }
    return 0;
}

int epoll_wait_debug =1;
static void *__ipc_stream_status_thrd ( void* arg )
{
    int nfds;
    int i;
	ep_data *epd;
	struct epoll_event events[MAX_EPOLL_FD];
    struct cha_priv_struct *hd_p = NULL;

	sigset_t set;
	sigemptyset(&set);
	sigaddset(&set, 54);

    prctl(PR_SET_NAME, "123" );
    pthread_detach(pthread_self());
    epollfd[0] = epoll_create(MAX_EPOLL_FD);


	while ( 1 ) {
        nfds = epoll_pwait(epollfd[0], &events[0], MAX_EPOLL_FD, EPOLL_TIMEOUT, &set);
		if ( epoll_wait_debug ) {
			xeLOG_NOTICE("epoll_wait return %d", nfds );
		}
		if ( nfds > 0 && nfds < MAX_EPOLL_FD ) {
			for ( i = 0; i < nfds; i++ ) {
				if ( !( events[i].events & EPOLLIN ) ) {
					continue;
				}
				usleep(20000);
				epd = (ep_data*)events[i].data.ptr;
                if(epd)
                    hd_p =  &dvr_cha_priv[epd->chid];
                produce_data(hd_p);
            }
        }
    }
	if ( epollfd[0] > 0 )
		close(epollfd[0]);
    return NULL;
}
int stream_main()
{
	pthread_t tid;
    pthread_create(&tid, NULL, __ipc_stream_status_thrd, NULL);
    sleep(1);
    tcp_stream_test_start("192.168.157.203");
    //sleep(2);
    STD_print("3");
    sleep(1);
    STD_print("2");
    sleep(1);
    STD_print("1");
    tcp_stream_test_close(&dvr_cha_priv[0]);
	getchar();
    return 0;
}
